package cn.itsource._05_rabbitmq_topic;

import cn.itsource.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

import static cn.itsource._04_rabbitmq_derict.send4.EXCHANGE_NAME_DERECT;
import static cn.itsource._05_rabbitmq_topic.send5.EXCHANGE_NAME_TOPIC;


/**
 * 1.消费者需要创建队列，是先创建队列然后再去绑定交换机
 * 2.消费者需要创建队列，但是使用的交换机是和发送者提供的交换机一样
 */
//消息接收者
public class rev5 {
    //定义一个队列
    public static final String QUEUE_FANOUT_TOPICS_02="queue_fanout_topics_02";
    public static void main(String[] args) throws Exception {
        //获得连接对象
        Connection connection = ConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //创建队列  durable：队列的持久化
        channel.queueDeclare(QUEUE_FANOUT_TOPICS_02, true, false, false, null);
        //绑定交换机
        channel.queueBind(QUEUE_FANOUT_TOPICS_02,EXCHANGE_NAME_TOPIC,"index.#");
        /**
         * 消费者接收消息调用此方法
         * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
         * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
        (收到消息失败后是否需要重新发送)
         * @param properties
         * @param body
         **/
        //设置每个消费者每次只能处理一条数据
        channel.basicQos(1);
        Consumer consume = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //消息的内容
                System.out.println(new String(body, "utf-8"));
                System.out.println("id:" + envelope.getDeliveryTag());
                System.out.println("队列名称:" + envelope.getRoutingKey());
                System.out.println("交换机:" + envelope.getExchange());

                //手动签收 long deliveryTag 消息的id, boolean multiple 是否等所有的消息都接收完再签收

                channel.basicAck(envelope.getDeliveryTag(), false);

            }
        };
        //接收消息
        //String basicConsume(String queue, DeliverCallback deliverCallback 接收方式, CancelCallback cancelCallback)
        channel.basicConsume(QUEUE_FANOUT_TOPICS_02, false, consume);
    }
}
